package com.northpool.operator.statistics.dataset;

import com.northpool.operator.annotation.InputParam;
import com.northpool.operator.annotation.InputParams;
import com.northpool.operator.annotation.OperatorInfo;
import com.northpool.operator.annotation.OutputParam;
import com.northpool.operator.annotation.OutputParams;
import com.northpool.operator.common.algorithm.OperatorTestBuild;
import com.northpool.operator.common.algorithm.TestParamBuilder;
import com.northpool.operator.common.stpclient.StpClientOperatorBase;
import com.northpool.operator.param.ParamTool;
import com.northpool.operator.param.custom.DatasetParam;
import com.northpool.operator.type.TypeEnum;
import com.northpool.stpclient.bean.BaseResponse;
import com.northpool.stpclient.modules.geoprocess.operator.IPipelineOperator;
import com.northpool.stpclient.operator.bean.DatasetBean;
import com.northpool.stpclient.operator.bean.OperatorBean;
import com.northpool.stpclient.operator.bean.ParamBean;
import com.northpool.stpclient.operator.bean.PipelineBean;

import java.io.File;
import java.util.LinkedList;
import java.util.List;

@OperatorInfo(
        ref = "SankeyStatisticsOperator",
        name = "桑基图统计",
        description = "生成Echart渲染所需的桑吉图数据"
)
@OutputParams({
        @OutputParam(ref = "result", name = "桑吉图结果", type = TypeEnum.STRING)
})
public class SankeyStatisticsOperator extends StpClientOperatorBase {

    @InputParams({
            @InputParam(ref = "dataset", name = "数据集", desc = "数据集包含数据源ID和数据表名称的JSON描述，如: {\"dataSourceId\":\"xxx-xxx\",\"tableName\":\"table\"}", type = TypeEnum.DB, required = true),
            @InputParam(ref = "sqlFilter", name = "SQL过滤条件", desc = "SQL过滤条件, 如: dlmc = '旱地'", type = TypeEnum.STRING, required = false),
            @InputParam(ref = "sourceField", name = "来源字段", desc = "来源字段名称，如：dlmc_1", type = TypeEnum.STRING, required = true),
            @InputParam(ref = "targetField", name = "目标字段", desc = "目标字段名称，如：dlmc_2", type = TypeEnum.STRING, required = true),
            @InputParam(ref = "valueField", name = "字段值", desc = "字段值名称，如：_area__", type = TypeEnum.STRING, required = true),
    })
    public SankeyStatisticsOperator() {
        super();
    }

    public static void main(String[] args) throws Exception {
        OperatorTestBuild.execute(SankeyStatisticsOperator.class,
                TestParamBuilder.aBuilder()
                        .param("dataset", "{\"dataSourceId\":\"402881ac934fb41701934fc212af000b\",\"tableName\":\"result.dltb_nmk_相交结果\"}")
                        .param("sqlFilter", "DLBM_1 not like '01%' and DLBM_2 like '01%'") //  非农用地 - > 农用地
//                        .param("sqlFilter", "DLBM_1 like '01%' and DLBM_2 not like '01%'") //  农用地 - > 非农用地
                        .param("sourceField", "DLMC_1")
                        .param("targetField", "DLMC_2")
                        .param("valueField", "_area__")
                        .build());
    }

    @Override
    public void execute(ParamTool params) throws Exception {
        DatasetParam dataset = params.getDataset("dataset");
        String dataSourceId = dataset.getDataSourceId();
        String tableName = dataset.getTableName();
        String sqlFilter = params.getStr("sqlFilter");

        String sourceField = params.getStr("sourceField");
        String targetField = params.getStr("targetField");
        String valueField = params.getStr("valueField");

        PipelineBean pipeline = new PipelineBean();
        pipeline.setResultFile(System.getProperty("user.dir") + File.separator + "result_file");
        DatasetBean dataSet = new DatasetBean();
        dataSet.setDataSourceId(dataSourceId);
        dataSet.setTableName(tableName);
        dataSet.setSqlFilter(sqlFilter);
        pipeline.setDataSet(dataSet);

        pipeline.setParallelism(16);

        List<OperatorBean> ops = new LinkedList<>();
        ops.add(buildSanKey(sourceField, targetField, valueField));
        pipeline.setResultFile(System.getProperty("user.dir") + File.separator + "result_file");
        pipeline.setOperators(ops);
        IPipelineOperator iPipelineOperator = this.client.pipelineOperator();
        BaseResponse response = iPipelineOperator.startPipeline(pipeline);
        iPipelineOperator.onStart(response.getRequestId(), () -> {
            this.log("面积统计开始");
        });
        iPipelineOperator.onSuccess(response.getRequestId(), () -> {
            this.log("面积统计成功");
            String log = iPipelineOperator.log(response.getRequestId());
            this.log(log);
            this.successExit();
        });
        iPipelineOperator.onError(response.getRequestId(), () -> {
            this.error("面积统计失败: {}", iPipelineOperator.log(response.getRequestId()));
            this.failedExit();
        });
        this.waiting(24);
    }

    public OperatorBean buildSanKey(String sourceField, String targetField, String valueField) {
        OperatorBean operatorBean = new OperatorBean();
        operatorBean.setClassName("com.northpool.stpprocess.operator.basic.statistics.SankeyOperator");

        List<ParamBean> params = new LinkedList<>();
        params.add(new ParamBean("sourceField", sourceField));
        params.add(new ParamBean("targetField", targetField));
        params.add(new ParamBean("valueField", valueField));
        operatorBean.setStartParam(params);

        return operatorBean;
    }
}
